package com.mjf.app.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.mjf.bean.TableProcess;
import com.mjf.common.GmallConfig;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.List;

/**
 * 分流 处理 广播流数据，主流数据（跟据广播流数据进行处理）
 */
public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {

    private Connection conn;
    private OutputTag<JSONObject> outputTag;
    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;

    public TableProcessFunction(OutputTag<JSONObject> outputTag, MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.outputTag = outputTag;
        this.mapStateDescriptor = mapStateDescriptor;
    }

    /**
     * 获取 phoenix 连接。
     * 如果担心主流数据比配置信息先来丢失数据，可以在open方法里面先将配置信息加载进来
     * @throws Exception
     */
    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName(GmallConfig.PHOENIX_DRIVER);
        conn = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
    }

    /**
     * 处理主流数据，通过配置信息将主流数据分流
     * @param value {"database":"", "tableName":"", "before":{}, "after":{}, "type":""}
     * @param ctx
     * @param out
     * @throws Exception
     */
    @Override
    public void processElement(JSONObject value, BroadcastProcessFunction<JSONObject, String, JSONObject>.ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        // 1.获取状态数据
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        String tableName = value.getString("tableName");
        String type = value.getString("type");
        String key = tableName + "-" + type;
        TableProcess tableProcess = broadcastState.get(key);

        if (tableProcess != null) {
            // 2.过滤字段
            JSONObject data = JSON.parseObject(value.getString("after"));
            filterColumn(data, tableProcess.getSinkColumns());
            value.put("after", data);   // 更新 after 数据

            // 3.分流
            // 将输出表/主题信息写入 value
            value.put("sinkTable", tableProcess.getSinkTable());

            String sinkType = tableProcess.getSinkType();
            if (TableProcess.SINK_TYPE_KAFKA.equals(sinkType)) {
                // Kafka 数据写入主流
                out.collect(value);
            } else if (TableProcess.SINK_TYPE_HBASE.equals(sinkType)) {
                // HBase 数据写入侧输出流
                ctx.output(outputTag, value);
            }
        } else {
            System.out.println("该组合Key: " + key + "在配置表中不存在！");
        }
    }

    /**
     * 将数据中不需要的列删除
     * @param data {"id":"11", "tm_name":"华为", "logo_url":"/fig/a.gif"}
     * @param sinkColumns id,tm_name
     */
    private void filterColumn(JSONObject data, String sinkColumns) {
        List<String> cols = Arrays.asList(sinkColumns.split(","));

/*        Iterator<Map.Entry<String, Object>> iterator = data.entrySet().iterator();
        while (iterator.hasNext()) {
            Map.Entry<String, Object> next = iterator.next();
            if (!cols.contains(next.getKey())) {
                iterator.remove();  // 将不需要的列数据删除
            }
        }*/

        // 简化形式
        data.entrySet().removeIf(next -> !cols.contains(next.getKey()));

    }

    /**
     * 处理广播流数据，实时加载配置信息
     * @param value {"database":"", "tableName":"", "before":{}, "after":{}, "type":""}
     * @param ctx
     * @param out
     * @throws Exception
     */
    @Override
    public void processBroadcastElement(String value, BroadcastProcessFunction<JSONObject, String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
        // 1.获取并解析数据
        JSONObject jsonObject = JSON.parseObject(value);
        String data = jsonObject.getString("after");
        TableProcess tableProcess = JSON.parseObject(data, TableProcess.class);

        // 2.建表(HBase需要手动建表)
        if (TableProcess.SINK_TYPE_HBASE.equals(tableProcess.getSinkType())) {
            createTable(
                    tableProcess.getSinkTable(),
                    tableProcess.getSinkColumns(),
                    tableProcess.getSinkPk(),
                    tableProcess.getSinkExtend()
            );
        }

        // 3.写入状态，广播出去
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        String key = tableProcess.getSourceTable() + "-" + tableProcess.getOperateType();
        broadcastState.put(key, tableProcess);
    }

    /**
     * HBase 建表语句
     * 建表语句：create table if not exists db.tn(id varchar primary key, tm_name varchar) xxx;
     * @param sinkTable   表名
     * @param sinkColumns 列名
     * @param sinkPk      主键
     * @param sinkExtend  扩展内容
     */
    private void createTable(String sinkTable, String sinkColumns, String sinkPk, String sinkExtend) {

        if (sinkPk == null) {
            sinkPk = "id";
        }

        if (sinkExtend == null) {
            sinkExtend = "";
        }

        StringBuffer createTableSQL = new StringBuffer("create table if not exists ")
                .append(GmallConfig.HBASE_SCHEMA)
                .append(".")
                .append(sinkTable)
                .append("(");

        String[] cols = sinkColumns.split(",");
        for (int i = 0; i < cols.length; i++) {
            String col = cols[i];

            if (sinkPk.equals(col)) {   // 如果是主键
                createTableSQL.append(col).append(" varchar primary key");
            } else {    // 如果不是主键
                createTableSQL.append(col).append(" varchar");
            }

            // 如果不是最后一个字段，拼接 ","
            if (i != cols.length - 1) {
                createTableSQL.append(",");
            }
        }

        createTableSQL.append(")").append(sinkExtend);

        // 打印建表语句
        System.out.println(createTableSQL);

        PreparedStatement preparedStatement = null;
        try {
            // 预编译 SQL
            preparedStatement = conn.prepareStatement(createTableSQL.toString());
            // 执行建表语句
            preparedStatement.execute();
        } catch (SQLException e) {
            throw new RuntimeException("HBase表[" + sinkTable + "]建表异常!");
        } finally {
            if (preparedStatement != null) {
                try {
                    preparedStatement.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }

    }
}
